%%--------------------------------------------------------------------
%% Copyright (c) 2013-2018 EMQ Enterprise, Inc. (http://emqtt.io)
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(quec_emq_kafka).

-include_lib("emqx/include/emqx.hrl").

-include_lib("brod/include/brod_int.hrl").

-export([load/1, unload/0]).

-define(APP, quec_emq_kafka).

-export([on_client_connected/4, on_client_disconnected/3]).
-export([on_client_subscribe/4, on_client_unsubscribe/4]).
%-export([on_session_created/3, on_session_subscribed/4, on_session_unsubscribed/4, on_session_terminated/4]).
%-export([on_session_subscribed/4, on_session_unsubscribed/4]).
%-export([on_message_publish/2, on_message_delivered/4, on_message_acked/4]).
-export([on_message_publish/2]).

-define(LOG(Level, Format, Args), lager:Level("kafka: " ++ Format, Args)).

%% Called when the plugin application start
load(Env) ->
  brod_init([Env]),
  emqx:hook('client.connected', fun ?MODULE:on_client_connected/4, [Env]),
  emqx:hook('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]),
  emqx:hook('client.subscribe', fun ?MODULE:on_client_subscribe/4, [Env]),
  emqx:hook('client.unsubscribe', fun ?MODULE:on_client_unsubscribe/4, [Env]),
  %emqx:hook('session.created', fun ?MODULE:on_session_created/3, [Env]),
  %emqx:hook('session.subscribed', fun ?MODULE:on_session_subscribed/4, [Env]),
  %emqx:hook('session.unsubscribed', fun ?MODULE:on_session_unsubscribed/4, [Env]),
  %emqx:hook('session.terminated', fun ?MODULE:on_session_terminated/4, [Env]),
  emqx:hook('message.publish', fun ?MODULE:on_message_publish/2, [Env]).
  %%emqx:hook('message.delivered', fun ?MODULE:on_message_delivered/4, [Env]).
  %emqx:hook('message.acked', fun ?MODULE:on_message_acked/4, [Env]).

on_client_connected(#{client_id := ClientId}, ConnAck, ConnAttrs,  _Env) ->
  io:format("client ~s connected, connack: ~w, conn_attrs:~p~n", [ClientId, ConnAck, ConnAttrs]),
  Event = mochijson2:encode([{type, <<"client_connected">>},
    {client_id, ClientId},
    {cluster_node, node()},
    {message, {}},
    {ts, emqx_time:now_ms()}]),  
  ok = produce_status(ClientId, Event),
  ok.

on_client_disconnected(#{client_id := ClientId, username := Username}, ReasonCode, _Env) ->
  io:format("Client(~s) disconnected, reason_code: ~w~n", [ClientId, ReasonCode]),
  Event =  mochijson2:encode([{type, <<"client_disconnected">>},
    {client_id, ClientId},
    {cluster_node, node()},
    {message, ReasonCode},
    {ts, emqx_time:now_ms()}]),  
  produce_status(ClientId, Event),
  ok.

on_client_subscribe(ClientId, Username, TopicTable, _Env) ->
  io:format("client(~s/~s) will subscribe: ~p~n", [Username, ClientId, TopicTable]),
  Event = mochijson2:encode([{type, <<"client_subscribe">>},
    {client_id, ClientId},
    {cluster_node, node()},
    {message, TopicTable},
    {ts, emqx_time:now_ms()}]),  
  ok = produce_status(ClientId, Event),
  {ok, TopicTable}.

on_client_unsubscribe(ClientId, Username, TopicTable, _Env) ->
  io:format("client(~s/~s) unsubscribe ~p~n", [ClientId, Username, TopicTable]),
  Event = mochijson2:encode([{type, <<"client_unsubscribe">>},
    {client_id, ClientId},
    {cluster_node, node()},
    {message, TopicTable},
    {ts, emqx_time:now_ms()}]),  
  ok = produce_status(ClientId, Event),
  {ok, TopicTable}.

on_session_created(ClientId, Username, _Env) ->
%%io:format("session(~s/~s) created.", [ClientId, Username]),
%% Json = mochijson2:encode([
%%   {type, <<"session_created">>},
%%   {client_id, ClientId},
%%   {user_name, Username},
%%   {cluster_node, node()},
%%   {ts, emqx_time:now_ms()}
%% ]),
%% %%ekaf_send_async(<<"emq_notify">>, Json),
  {ok, Username}.

on_session_subscribed(ClientId, Username, {Topic, Opts}, _Env) ->
  {ok, {Topic, Opts}}.

on_session_unsubscribed(ClientId, Username, {Topic, Opts}, _Env) ->
  ok.

on_session_terminated(ClientId, Username, Reason, _Env) ->
  io:format("session(~s/~s) terminated: ~p.~n", [ClientId, Username, Reason]),
  stop.

on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
    % io:format("message publish: ~p.", [topic]),
    {ok, Message};
on_message_publish(Message = #message{id = MsgId,
                        qos = Qos,
                        from = From,
                        flags = Flags,
                        topic  = Topic,
                        payload = Payload,
                        timestamp  = Time
						}, _Env) -> 
    lager:info("Publish ~s~n", [emqx_message:format(Message)]),
    Json = mochijson2:encode([
        {type, <<"published">>},
        {client_id, From},
        {message, [
%%          {username, Username},
          {topic, Topic},
          {payload, base64:encode_to_string(Payload)},
          {qos, Qos}
%%          {dup, Dup},
%%          {retain, Retain}
        ]},
        {cluster_node, node()},
        {ts, emqx_time:now_ms(Time)}
    ]),
    ok = produce_points(From, Json),
    {ok, Message}.

on_message_delivered(ClientId, Username, Message, _Env) ->
  io:format("delivered to client(~s/~s): ~s~n", [Username, ClientId, emqx_message:format(Message)]),
  {ok, Message}.

on_message_acked(ClientId, Username, Message, _Env) ->
  io:format("client(~s/~s) acked: ~s~n", [Username, ClientId, emqx_message:format(Message)]),
  {ok, Message}.


brod_init(_Env) ->
    {ok, BootstrapBrokers} = get_bootstrap_brokers(),
%    {ok, DpTopic, _, _} = get_points_topic(),
%    {ok, DsTopic, _, _} = get_status_topic(),
    ok = brod:start(),
    ClientConfig =
        [
           {reconnect_cool_down_seconds, 10},
           {auto_start_producers, true},
           {default_producer_config, []},
           {allow_topic_auto_creation, true}
        ],

    ok = brod:start_client(BootstrapBrokers, brod_client_1, ClientConfig),
    lager:info("Init brod kafka client with ~p", [BootstrapBrokers]).

produce_points(ClientId, Json) ->
    Topic = get_points_topic(),
    produce(Topic, ClientId, Json),
    ok.

produce_status(ClientId, Json) ->
    Topic = get_status_topic(),
    produce(Topic, ClientId, Json),
    ok.

produce(TopicInfo, ClientId, Json) ->
    case TopicInfo of
        {ok, Topic, custom, _}->
            brod_produce(Topic, hash, ClientId, Json);
        {ok, Topic, _, _} ->
            brod_produce(Topic, random, ClientId, Json)
    end.

brod_produce(Topic, Partitioner, ClientId, Json) ->
    {ok, CallRef} = brod:produce(brod_client_1, Topic, Partitioner, ClientId, list_to_binary(Json)),
    receive
        #brod_produce_reply{call_ref = CallRef, result = brod_produce_req_acked} -> ok
    after 5000 ->
        lager:error("Produce message to ~p for ~p timeout.",[Topic, ClientId])
    end,
    ok.


get_bootstrap_brokers() ->
    application:get_env(?APP, bootstrap_brokers).

get_config_prop_list() ->
    application:get_env(?APP, config).

get_instrument_config() ->
    {ok, Values} = get_config_prop_list(),
    Instrument = proplists:get_value(instrument, Values),
    {ok, Instrument}.

%% 从配置中获取设备数据流主题Points的配置
get_points_topic() ->
    {ok, Values} = application:get_env(?APP, points),
    get_topic(Values).

%% 从配置中获取设备状态流主题Status的配置
get_status_topic() ->
    {ok, Values} = application:get_env(?APP, status),
    get_topic(Values).

get_topic(Values) ->
    Topic = proplists:get_value(topic, Values),
    PartitionStrategy = proplists:get_value(partition_strategy, Values),
    PartitionWorkers = proplists:get_value(partition_workers, Values),
    {ok, Topic, PartitionStrategy, PartitionWorkers}.


%% Called when the plugin application stop
unload() ->
  emqx:unhook('client.connected', fun ?MODULE:on_client_connected/4),
  emqx:unhook('client.disconnected', fun ?MODULE:on_client_disconnected/3),
  emqx:unhook('client.subscribe', fun ?MODULE:on_client_subscribe/4),
  emqx:unhook('client.unsubscribe', fun ?MODULE:on_client_unsubscribe/4),
%  emqx:unhook('session.created', fun ?MODULE:on_session_created/3),
%  emqx:unhook('session.subscribed', fun ?MODULE:on_session_subscribed/4),
%  emqx:unhook('session.unsubscribed', fun ?MODULE:on_session_unsubscribed/4),
%  emqx:unhook('session.terminated', fun ?MODULE:on_session_terminated/4),
  emqx:unhook('message.publish', fun ?MODULE:on_message_publish/2),
%  emqx:unhook('message.delivered', fun ?MODULE:on_message_delivered/4),
%  emqx:unhook('message.acked', fun ?MODULE:on_message_acked/4).
  brod:stop_client(brod_client_1).

